package org.example;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class Consumer {

    private KafkaConsumer<String, String> kafkaConsumer;


    /*
     * 初始化
     * */
    Consumer(String topic) {

        //创建配置对象
        Properties properties = new Properties();
        /*
         * 进行配置
         * */
        //此处为虚拟机ip
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.222:9092");

        //本地ip
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        //配置序列化--必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());



        //配置消费者组--必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "device");

        //创建消费者对象
        kafkaConsumer = new KafkaConsumer<String, String>(properties);

        //注册要消费的主题(单个消费者消费整个主题数据)
        ArrayList<String> topicset = new ArrayList<>();
        topicset.add(topic);
        kafkaConsumer.subscribe(topicset);

        //消费某个主题的某个分区数据
//        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
//        topicPartitions.add(new TopicPartition(topic,0));
//        topicPartitions.add(new TopicPartition(topic,1));
//        topicPartitions.add(new TopicPartition(topic,2));
//        topicPartitions.add(new TopicPartition(topic,3));
//        topicPartitions.add(new TopicPartition(topic,4));
//        kafkaConsumer.assign(topicPartitions);

    }

    /*
    * 消费数据
    * */
    public void mysql_consume(){

        /*
        * 初始化mysql
        * */

        MysqlConn mysqlConn = new MysqlConn();


        //拉取数据打印
        while (true) {

            //设置1s消费一批数据
            ConsumerRecords<String, String> consumerRecord = kafkaConsumer.poll(Duration.ofSeconds(1));

            //打印消费到的数据
            for (ConsumerRecord<String, String> record : consumerRecord) {
//                System.out.println(record.partition()+"  "+record.offset());
                System.out.println(record);
                //消费到mysql
                mysqlConn.add(record.timestamp(),record.value(),record.partition());
            }
        }
    }

}
